package com.dyson.concurrent.audition.demo05;

import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 生产者消费者模式
 * 使用线程安全的队列实现
 * <p>
 * 证明阻塞队列和并发队列的线程安全，
 * 指的是对该对象的某一个方法执行过程中是线程安全，
 * 但是对于其它附加的逻辑是不能保证线程安全的。
 * <p>
 * 使用非阻塞方法只能保证方法的原子性，但是由于队列大小为0时
 * 没有进行等待，所以最后结果是不能达到预期结果的。
 *
 * @author Peter
 */
public class ProducerAndConsumerTest01 {

    private static Queue<Integer> blcQ = new ArrayBlockingQueue<>(100_0000);

//	private static Queue<Integer> blcQ = new LinkedBlockingQueue<>();

    private static Queue<Integer> cocrtQ = new ConcurrentLinkedQueue<>();

    public static void main(String[] args) throws InterruptedException {
        // 生产者线程
//		Producer p1 = new Producer(blcQ);
//		Producer p2 = new Producer(blcQ);
//		Producer p3 = new Producer(blcQ);
//		Consumer c1 = new Consumer(blcQ);
//		Consumer c2 = new Consumer(blcQ);

        Producer p1 = new Producer(cocrtQ);
        Producer p2 = new Producer(cocrtQ);
        Producer p3 = new Producer(cocrtQ);
        Consumer c1 = new Consumer(cocrtQ);
        Consumer c2 = new Consumer(cocrtQ);

        p1.start();
        c1.start();
        p2.start();
        p3.start();
        c2.start();


        p1.join();
        p2.join();
        p3.join();
        c1.join();
        c2.join();
//		System.out.println("最后的大小是："+blcQ.size());
        System.out.println("最后的大小是：" + cocrtQ.size());

    }

}

class Producer extends Thread {

    private Queue<Integer> q;

    public Producer(Queue<Integer> q) {
        this.q = q;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10_00; i++) {
            int randomNum = (int) (Math.random() * 5_00 + 1);
            q.add(randomNum);
        }
    }
}


class Consumer extends Thread {

    private Queue<Integer> q;

    public Consumer(Queue<Integer> q) {
        this.q = q;
    }

    @Override
    public void run() {
        while (q.size() > 0) {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            q.remove();
//			q.poll();
        }


        for (int i = 0; i < 10_0000; i++) {
			q.remove();//队列为空后直接抛异常了
            //q.poll();//队列为空后直接返回null，但是本次操作会成立
			try {
				((BlockingQueue<Integer>)q).take();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
        }
    }
}
